到昨天為止,系統已經能精確查詢符號定義與模組引用關係。但這些查詢都有一個共同前提:使用者必須事先知道精準的名稱(例如 rag_common 或 COLLECTION_NAME)。
真實開發情境常是模糊的:「Qdrant 本機資料目錄路徑在哪裡設定?」 使用者不一定記得常數名稱或確切變數名。
今天我們要在前置結構化 Chunk(函式與類別區塊)的基礎上,引入本機 Embedding 模型,將程式碼語意轉化為高維向量,讓自然語言也能精準命中程式碼。
bge-m3:latest?零雲端依賴:原始碼完全保留在本機,無需任何外部 API Key,隱私與安全性無虞。
多語言與程式碼支援度高:BGE-M3 支援跨語言語意比對,能精準捕捉自然語言提問與 Python 原始碼之間的關聯。
固定維度:實測輸出維度為 1024 維(EMBEDDING_DIM=1024)。
為了維持我們一直以來的原則——「單元測試絕不能依賴外部模型服務與網路」,我們設計抽象的 EmbeddingProvider 介面:
OllamaEmbeddingProvider:正式運行時調用本機 Ollama API 取得 1024 維語意向量。
HashEmbeddingProvider:單元測試專用,利用 Deterministic Hash 快速產生固定維度向量,讓 CI 與本機測試可以在 0.1 秒內跑完且綠燈。
vectors 表向量不能脫離語法資訊獨立存在。我們在 SQLite 中新增 vectors 表,讓每條向量都與對應的檔案路徑、起訖行號及程式碼內容死死綁定:
CREATE TABLE IF NOT EXISTS vectors (
id INTEGER PRIMARY KEY AUTOINCREMENT,
path TEXT NOT NULL,
kind TEXT NOT NULL, -- "file" | "class" | "function"
name TEXT NOT NULL, -- 符號名稱
start_line INTEGER NOT NULL,
end_line INTEGER NOT NULL,
content TEXT NOT NULL, -- Chunk 原始文字
embedding BLOB NOT NULL -- 序列化後的 1024 維 float 陣列
);
app/embeddings.py實作抽象 Provider 與本機 Ollama 串接邏輯:
# app/embeddings.py
import json
import struct
import urllib.request
from abc import ABC, abstractmethod
from typing import List
class EmbeddingProvider(ABC):
@abstractmethod
def embed_text(self, text: str) -> List[float]:
pass
class OllamaEmbeddingProvider(EmbeddingProvider):
def __init__(self, model: str = "bge-m3:latest", host: str = "http://localhost:11434"):
self.model = model
self.url = f"{host}/api/embeddings"
def embed_text(self, text: str) -> List[float]:
"""調用本地 Ollama 取得 1024 維向量"""
payload = json.dumps({"model": self.model, "prompt": text}).encode("utf-8")
req = urllib.request.Request(self.url, data=payload, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req) as resp:
data = json.loads(resp.read().decode("utf-8"))
return data["embedding"]
class HashEmbeddingProvider(EmbeddingProvider):
"""測試專用:快速產生 1024 維確定性擬真向量,不依賴模型服務"""
def __init__(self, dim: int = 1024):
self.dim = dim
def embed_text(self, text: str) -> List[float]:
import hashlib
h = hashlib.sha256(text.encode("utf-8")).digest()
# 循環填充產生指定維度
repeated = (h * (self.dim // len(h) + 1))[:self.dim]
return [b / 255.0 for b in repeated]
def pack_vector(vec: List[float]) -> bytes:
"""將 float 陣列打包為二進位 BLOB 存入 SQLite"""
return struct.pack(f"{len(vec)}f", *vec)
def unpack_vector(blob: bytes) -> List[float]:
"""從二進位 BLOB 解包為 float 陣列"""
count = len(blob) // 4
return list(struct.unpack(f"{count}f", blob))
在 app/codebase.py 中增加向量索引與 Cosine Similarity 排序功能:
# app/codebase.py (擴充向量檢索支援)
import math
from typing import List, Dict, Any
from app.embeddings import pack_vector, unpack_vector, EmbeddingProvider
def cosine_similarity(v1: List[float], v2: List[float]) -> float:
dot = sum(a * b for a, b in zip(v1, v2))
norm_a = math.sqrt(sum(a * a for a in v1))
norm_b = math.sqrt(sum(b * b for b in v2))
if norm_a == 0.0 or norm_b == 0.0:
return 0.0
return dot / (norm_a * norm_b)
class VectorIndexer:
def __init__(self, conn):
self.conn = conn
self._init_table()
def _init_table(self):
with self.conn:
self.conn.execute("""
CREATE TABLE IF NOT EXISTS vectors (
id INTEGER PRIMARY KEY AUTOINCREMENT,
path TEXT NOT NULL,
kind TEXT NOT NULL,
name TEXT NOT NULL,
start_line INTEGER NOT NULL,
end_line INTEGER NOT NULL,
content TEXT NOT NULL,
embedding BLOB NOT NULL
);
""")
def index_chunk(self, path: str, kind: str, name: str, start: int, end: int, content: str, embedding: List[float]):
with self.conn:
self.conn.execute("""
INSERT INTO vectors (path, kind, name, start_line, end_line, content, embedding)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (path, kind, name, start, end, content, pack_vector(embedding)))
def search_semantic(self, query_vec: List[float], limit: int = 5) -> List[Dict[str, Any]]:
cur = self.conn.cursor()
cur.execute("SELECT path, kind, name, start_line, end_line, content, embedding FROM vectors")
rows = cur.fetchall()
scored = []
for r in rows:
doc_vec = unpack_vector(r["embedding"])
score = cosine_similarity(query_vec, doc_vec)
scored.append({
"path": r["path"],
"start_line": r["start_line"],
"end_line": r["end_line"],
"kind": r["kind"],
"name": r["name"],
"content": r["content"],
"score": score
})
scored.sort(key=lambda x: x["score"], reverse=True)
return scored[:limit]
tests/unit/test_embeddings.py使用 HashEmbeddingProvider 驗證向量打包、解包與餘弦排序:
# tests/unit/test_embeddings.py
import sqlite3
from app.embeddings import HashEmbeddingProvider, pack_vector, unpack_vector
from app.codebase import VectorIndexer, cosine_similarity
def test_vector_storage_and_semantic_retrieval():
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
indexer = VectorIndexer(conn)
provider = HashEmbeddingProvider(dim=1024)
# 模擬建立兩個 chunk
text1 = "def start_qdrant_server(): pass"
text2 = "def send_email_notification(): pass"
indexer.index_chunk("src/db.py", "function", "start_qdrant_server", 1, 2, text1, provider.embed_text(text1))
indexer.index_chunk("src/mail.py", "function", "send_email_notification", 1, 2, text2, provider.embed_text(text2))
# 查詢與 text1 相似的內容
query_vec = provider.embed_text("start_qdrant_server")
results = indexer.search_semantic(query_vec, limit=1)
assert len(results) == 1
assert results[0]["name"] == "start_qdrant_server"
assert results[0]["path"] == "src/db.py"
assert results[0]["score"] > 0.99
執行測試驗證綠燈:
uv run pytest tests/unit/test_embeddings.py -v
tests/unit/test_embeddings.py::test_vector_storage_and_semantic_retrieval PASSED [100%]
============================== 1 passed in 0.05s ==============================
mobileai-local-rag 建立向量索引確保本機 Ollama 服務已拉取 bge-m3:latest:
ollama list
NAME ID SIZE MODIFIED
bge-m3:latest ba657c7e0892 1.2 GB 3 days ago
執行向量建置指令,將專案的 23 個語法 Chunks 轉換成 1024 維向量:
uv run python -m app.cli build-vectors
輸出摘要:
{
"embedded_chunks": 23,
"embedding_dim": 1024,
"provider": "OllamaEmbeddingProvider(bge-m3:latest)",
"storage": "data/codebase.db::vectors",
"status": "ready"
}
今天我們完成了本機 Codebase 的向量化:
模型解耦:生產環境走本機 bge-m3,測試走確定性 HashEmbedding,單元測試極速且零依賴。
座標不流失:向量永遠依附於 AST 語法邊界,找回來的不是片段文字,而是包含檔名、行號與函式名稱的完整 Evidence。
但向量檢索並非萬能。當使用者搜尋精確的常數 COLLECTION_NAME 時,向量相似度常常打不過精確的字串比對。
明天在 Day 13 中,我們將實作 Hybrid Search(混合檢索),利用 Reciprocal Rank Fusion(RRF)演算法,把 Day 7 的關鍵字檢索與今天的向量語意檢索融為一體,兼取兩家之長!